package io.github.rascaler.dcenter.manager;

import com.alibaba.fastjson.JSON;
import com.binance.connector.futures.client.exceptions.BinanceClientException;
import com.binance.connector.futures.client.exceptions.BinanceConnectorException;
import com.binance.connector.futures.client.impl.UMFuturesClientImpl;
import com.binance.connector.futures.client.utils.ProxyAuth;
import io.github.rascaler.dcenter.constant.ConstantItem;
import io.github.rascaler.dcenter.db.DbManager;
import io.github.rascaler.dcenter.platforms.DbMapper;
import io.github.rascaler.dcenter.platforms.Frequency;
import io.github.rascaler.dcenter.platforms.Parser;
import io.github.rascaler.dcenter.platforms.binance.BinanceCandlestickParser;
import io.github.rascaler.dcenter.platforms.binance.BinanceFrequency;
import io.github.rascaler.dcenter.platforms.binance.BinanceInstrumentParser;
import io.github.rascaler.dcenter.pojo.Candlestick;
import io.github.rascaler.dcenter.pojo.FixedArrayList;
import io.github.rascaler.dcenter.pojo.InitKline;
import io.github.rascaler.dcenter.pojo.Instrument;
import io.github.rascaler.dcenter.properties.BinanceProperties;
import io.github.rascaler.dcenter.properties.ProxyInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.net.InetSocketAddress;
import java.net.Proxy;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

@Slf4j
@Component
public class BinanceManager {

    @Autowired
    private DbManager dbManager;

    @Autowired
    private BinanceProperties binanceProperties;

    public final static UMFuturesClientImpl client = new UMFuturesClientImpl();

    @Resource(type = BinanceCandlestickParser.class)
    private Parser binanceCandlestickParser;

    @Resource(type = BinanceInstrumentParser.class)
    private Parser binanceInstrumentParser;

    @Autowired
    private DbMapper dbMapper;

    // 经过测试请求每分钟频率不能超过480次，下一分钟会重置频率，哪怕你上一分钟在59秒的时候请求480次
    // 分配400频次给1分钟采集，60频次给上架下架，20频次预留，不能卡太死
    public final static AtomicLong totalLimit = new AtomicLong(0);

    public final static int MAX_LIMIT = 400;

    // 实时最大频率
    public final static int REAL_MAX_LIMIT = 400;

    // 上/下架最大频率
    public final static int ON_OFF_MAX_LIMIT = 60;



    public void init() {
        if (dbManager.getState() == 1) {
            return;
        }

        if (binanceProperties.getUseProxy() && null == client.getProxy()) {
            // 设置代理
            ProxyInfo proxySetting = binanceProperties.getProxies().get(0);
            Proxy proxyConn = new Proxy(Proxy.Type.HTTP, new InetSocketAddress(proxySetting.getHost(), proxySetting.getPort()));
            ProxyAuth proxy = new ProxyAuth(proxyConn, null);
            client.setProxy(proxy);
        }

        // 采集instruments
        String instStr = client.market().exchangeInfo();
        List<Instrument> instruments = binanceInstrumentParser.parse(instStr);
        dbManager.getMdb().setInstrumentMap(instruments.stream().collect(Collectors.toMap(Instrument::getSymbol, i -> i)));
        // bar
        Set<String> frequencies = new HashSet<>(binanceProperties.getFrequencies());
        // 将1分钟进行最后初始化
        frequencies.remove(ConstantItem.FREQUENCY_1MIN);
        frequencies.add(ConstantItem.FREQUENCY_1MIN);
        List<String> needInitSymbols = dbMapper.getNeedInitSymbols(ConstantItem.FREQUENCY_1MIN);
        // 判断是否有未初始的数据
        List<InitKline> waitingInitKlines = new ArrayList<>();
        if (needInitSymbols.size() > 0) {
            log.error("需要初始化的币:{}", JSON.toJSONString(needInitSymbols));
            binanceProperties.getFrequencies().forEach(f -> {
                needInitSymbols.forEach(s -> {
                        InitKline initKline = new InitKline();
                        initKline.setSymbol(s);
                        initKline.setFrequency(f);
                        initKline.setAdjust(ConstantItem.ADJUST_NO);
                        initKline.setExchange(ConstantItem.EXCHANGE_BINANCE);
                        waitingInitKlines.add(initKline);
                    });
            });
        }

        if (waitingInitKlines.size() == 0) {
            if (dbManager.getState() == 0) {
                dbManager.setState(2);
            }
            // 去掉下架的币
            List<String> symbols = instruments.stream().map(Instrument::getSymbol).collect(Collectors.toList());
            List<String> offSymbols = dbManager.getMdb().getBarMap().keySet()
                    .stream().filter(k -> !symbols.contains(k.split(":")[1])).collect(Collectors.toList());
            offSymbols.forEach(o -> {
                log.error("下架:{}", o);
                dbManager.getMdb().getBarMap().remove(o);
            });
            return;
        }

        if (dbManager.getState() == 1) {
            return;
        }
        dbManager.setState(1);
        ExecutorService pool = Executors.newFixedThreadPool(waitingInitKlines.size() > 100 ? 100 : waitingInitKlines.size());
        List<Future<String>> tasks = new ArrayList<>();
        List<String> result = new ArrayList<>();
        int total = waitingInitKlines.size();
        int successCount = 0;
        while (waitingInitKlines.size() > 0) {
            // 防止这一分钟已经出现过采集，导致大批量的采集失败
            if (total > 60) {
                try {
                    Thread.sleep(60000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            result.clear();
            tasks.clear();
            List<InitKline> maxInit = waitingInitKlines.size() < binanceProperties.getInitMinuteLimt() ? waitingInitKlines : waitingInitKlines.subList(0, binanceProperties.getInitMinuteLimt());
            for (InitKline i : maxInit) {
                Future<String> future = pool.submit(() -> {
                    LinkedHashMap<String, Object> parameters = new LinkedHashMap<>();
                    parameters.put("symbol", i.getSymbol());
                    parameters.put("interval", BinanceFrequency.getFrequency(i.getFrequency()));
                    parameters.put("limit", binanceProperties.getInitBarLength());

                    try {
                        String data = client.market().klines(parameters);
                        return data;
                    } catch (BinanceConnectorException e) {
                        log.error(String.format("接口请求失败：【%s-%s】",i.getSymbol(), i.getFrequency()), e);
                        log.error("fullErrMessage: {}", e.getMessage(), e);
                    } catch (BinanceClientException e) {
                        log.error(String.format("接口请求失败：【%s-%s】", i.getSymbol(), i.getFrequency()), e);
                        log.error("fullErrMessage: {} \nerrMessage: {} \nerrCode: {} \nHTTPStatusCode: {}",
                                e.getMessage(), e.getErrMsg(), e.getErrorCode(), e.getHttpStatusCode(), e);
                    } catch (Exception e) {

                    }
                    return null;
                });
                tasks.add(future);
            }

            long startTime = System.currentTimeMillis();
            // 获取结果
            for (int i = 0;i < maxInit.size();i++) {
                Future<String> t = tasks.get(i);
                try {
                    result.add(t.get());
                } catch (Exception e) {
                    InitKline current = maxInit.get(i);
                    log.error(String.format("获取结果集失败：【%s-%s】", current.getSymbol(), current.getFrequency()), e);
                }
            }
            int currentSuccessCount = result.stream().filter(r -> null != r).collect(Collectors.toList()).size();
            successCount += currentSuccessCount;
            log.error("数据采集完成，目标数：{}，成功数：{}，失败数：{}，耗时：{}ms,当前进度{}%",
                    maxInit.size(),
                    currentSuccessCount,
                    result.stream().filter(r -> null == r).collect(Collectors.toList()).size(),
                    System.currentTimeMillis() - startTime,
                    new BigDecimal(successCount*100).divide(new BigDecimal(total),RoundingMode.HALF_UP).doubleValue());

            startTime = System.currentTimeMillis();
            // 保存并写入时间
            for (int i = 0;i < maxInit.size();i++) {
                InitKline s = maxInit.get(i);
                String key = String.format("binance:%s:%s:%s:%s", s.getSymbol(), s.getFrequency(), s.getAdjust(), s.getExchange());
                // OHLCV
                if (StringUtils.isBlank(result.get(i))) {
                    continue;
                }
                List<Candlestick> candlesticks = binanceCandlestickParser.parse(result.get(i));
                FixedArrayList<Candlestick> bars = dbManager.getMdb().getBars(key);
                bars.clear();
                bars.addAll(candlesticks);
            }

            // 移除已经成功的任务
            List<InitKline> successList = new ArrayList<>();
            for (int i = 0; i < result.size();i++) {
                if (null != result.get(i)) {
                    successList.add(maxInit.get(i));
                }
            }
            waitingInitKlines.removeAll(successList);

            // 防止直接
            if (total > 60 && waitingInitKlines.size() == 0) {
                try {
                    Thread.sleep(60000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

        // 保存
        dbManager.setState(2);
        dbManager.getMdb().saveDatas();
        pool.shutdown();
    }
}
